package com.reactim.message.service.impl;

import com.reactim.common.util.IdGenerator;
import com.reactim.message.dto.MessageResponse;
import com.reactim.message.dto.SendMessageRequest;
import com.reactim.message.entity.Message;
import com.reactim.message.entity.MessageReadStatus;
import com.reactim.message.repository.MessageReadStatusRepository;
import com.reactim.message.repository.MessageRepository;
import com.reactim.message.service.ConversationService;
import com.reactim.message.service.MessageService;
import com.reactim.message.service.MessagePushService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 消息服务实现类
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class MessageServiceImpl implements MessageService {
    
    private final MessageRepository messageRepository;
    private final MessageReadStatusRepository messageReadStatusRepository;
    private final ConversationService conversationService;
    private final MessagePushService messagePushService;
    
    // 消息序号生成器（简化实现，实际应该使用分布式序号生成）
    private final AtomicLong sequenceGenerator = new AtomicLong(0);
    
    @Override
    public Mono<MessageResponse> sendMessage(SendMessageRequest request) {
        return Mono.fromCallable(() -> {
            // 创建消息实体
            Message message = new Message();
            message.setMessageId(IdGenerator.generateMessageId());
            message.setFromUserId(request.getFromUserId());
            message.setToUserId(request.getToUserId());
            message.setGroupId(request.getGroupId());
            message.setMessageType(request.getMessageType());
            message.setContent(request.getContent());
            message.setExtra(request.getExtra());
            message.setStatus(Message.MessageStatus.NORMAL);
            message.setCreatedAt(LocalDateTime.now());
            message.setUpdatedAt(LocalDateTime.now());
            message.setSequence(sequenceGenerator.incrementAndGet());
            
            // 生成会话ID
            String conversationId;
            if (request.getGroupId() != null) {
                // 群聊
                conversationId = conversationService.generateConversationId(
                    com.reactim.message.entity.Conversation.ConversationType.GROUP, 
                    request.getGroupId()
                );
            } else {
                // 私聊
                conversationId = conversationService.generateConversationId(
                    com.reactim.message.entity.Conversation.ConversationType.PRIVATE,
                    request.getFromUserId(), 
                    request.getToUserId()
                );
            }
            message.setConversationId(conversationId);
            
            return message;
        })
        .flatMap(messageRepository::save)
        .flatMap(savedMessage -> {
            // 更新会话最后消息信息
            String messagePreview = generateMessagePreview(savedMessage);
            return conversationService.updateLastMessage(
                savedMessage.getConversationId(),
                savedMessage.getMessageId(),
                messagePreview
            ).then(Mono.just(savedMessage));
        })
        .map(this::convertToMessageResponse)
        .flatMap(messageResponse -> {
            // 推送消息给接收方
            return pushMessageToReceivers(messageResponse)
                    .then(Mono.just(messageResponse));
        })
        .doOnSuccess(response -> log.info("消息发送成功: {}", response.getMessageId()))
        .doOnError(error -> log.error("消息发送失败", error));
    }
    
    @Override
    public Mono<MessageResponse> getMessageById(String messageId) {
        return messageRepository.findByMessageId(messageId)
                .map(this::convertToMessageResponse)
                .switchIfEmpty(Mono.error(new RuntimeException("消息不存在")));
    }
    
    @Override
    public Flux<MessageResponse> getConversationMessages(String conversationId, Pageable pageable) {
        return messageRepository.findByConversationIdOrderByCreatedAtDesc(conversationId, pageable)
                .map(this::convertToMessageResponse);
    }
    
    @Override
    public Flux<MessageResponse> getPrivateMessages(Long userId1, Long userId2, Pageable pageable) {
        return messageRepository.findPrivateMessages(userId1, userId2, pageable)
                .map(this::convertToMessageResponse);
    }
    
    @Override
    public Flux<MessageResponse> getGroupMessages(Long groupId, Pageable pageable) {
        return messageRepository.findByGroupIdAndStatusOrderByCreatedAtDesc(
                groupId, Message.MessageStatus.NORMAL, pageable)
                .map(this::convertToMessageResponse);
    }
    
    @Override
    public Mono<MessageResponse> recallMessage(String messageId, Long userId) {
        return messageRepository.findRecallableMessage(messageId, userId)
                .switchIfEmpty(Mono.error(new RuntimeException("消息不存在或无权撤回")))
                .flatMap(message -> {
                    // 检查撤回时间限制（例如：2分钟内可撤回）
                    if (message.getCreatedAt().isBefore(LocalDateTime.now().minusMinutes(2))) {
                        return Mono.error(new RuntimeException("消息发送时间过长，无法撤回"));
                    }
                    
                    message.setStatus(Message.MessageStatus.RECALLED);
                    message.setUpdatedAt(LocalDateTime.now());
                    return messageRepository.save(message);
                })
                .map(this::convertToMessageResponse)
                .doOnSuccess(response -> log.info("消息撤回成功: {}", messageId));
    }
    
    @Override
    public Mono<Void> deleteMessage(String messageId, Long userId) {
        return messageRepository.findByMessageId(messageId)
                .switchIfEmpty(Mono.error(new RuntimeException("消息不存在")))
                .flatMap(message -> {
                    // 检查删除权限
                    if (!message.getFromUserId().equals(userId)) {
                        return Mono.error(new RuntimeException("无权删除此消息"));
                    }
                    
                    message.setStatus(Message.MessageStatus.DELETED);
                    message.setUpdatedAt(LocalDateTime.now());
                    return messageRepository.save(message);
                })
                .then()
                .doOnSuccess(v -> log.info("消息删除成功: {}", messageId));
    }
    
    @Override
    public Mono<MessageResponse> editMessage(String messageId, Long userId, Message.MessageContent newContent) {
        return messageRepository.findByMessageId(messageId)
                .switchIfEmpty(Mono.error(new RuntimeException("消息不存在")))
                .flatMap(message -> {
                    // 检查编辑权限
                    if (!message.getFromUserId().equals(userId)) {
                        return Mono.error(new RuntimeException("无权编辑此消息"));
                    }
                    
                    // 检查消息类型是否支持编辑
                    if (message.getMessageType() != Message.MessageType.TEXT) {
                        return Mono.error(new RuntimeException("此类型消息不支持编辑"));
                    }
                    
                    // 保存编辑历史
                    if (message.getExtra() == null) {
                        message.setExtra(new Message.MessageExtra());
                    }
                    if (message.getExtra().getEditInfo() == null) {
                        message.getExtra().setEditInfo(new Message.EditInfo());
                        message.getExtra().getEditInfo().setEditCount(0);
                    }
                    
                    // 更新编辑信息
                    Message.EditInfo editInfo = message.getExtra().getEditInfo();
                    editInfo.setOriginalContent(message.getContent().toString());
                    editInfo.setEditTimestamp(LocalDateTime.now());
                    editInfo.setEditCount(editInfo.getEditCount() + 1);
                    
                    // 更新消息内容
                    message.setContent(newContent);
                    message.setStatus(Message.MessageStatus.EDITED);
                    message.setUpdatedAt(LocalDateTime.now());
                    
                    return messageRepository.save(message);
                })
                .map(this::convertToMessageResponse)
                .doOnSuccess(response -> log.info("消息编辑成功: {}", messageId));
    }
    
    @Override
    public Mono<Void> markMessageAsRead(String messageId, Long userId) {
        return messageReadStatusRepository.findByMessageId(messageId)
                .switchIfEmpty(createNewReadStatus(messageId))
                .flatMap(readStatus -> {
                    // 检查是否已经标记为已读
                    boolean alreadyRead = readStatus.getReadStatus().stream()
                            .anyMatch(info -> info.getUserId().equals(userId));
                    
                    if (!alreadyRead) {
                        MessageReadStatus.ReadInfo readInfo = new MessageReadStatus.ReadInfo();
                        readInfo.setUserId(userId);
                        readInfo.setReadAt(LocalDateTime.now());
                        readStatus.getReadStatus().add(readInfo);
                        
                        return messageReadStatusRepository.save(readStatus);
                    }
                    
                    return Mono.just(readStatus);
                })
                .then()
                .doOnSuccess(v -> log.debug("消息标记已读: {} by user: {}", messageId, userId));
    }
    
    @Override
    public Mono<Void> markMessagesAsRead(List<String> messageIds, Long userId) {
        return Flux.fromIterable(messageIds)
                .flatMap(messageId -> markMessageAsRead(messageId, userId))
                .then();
    }
    
    @Override
    public Mono<Long> getUnreadMessageCount(Long userId) {
        return messageRepository.countUnreadMessages(userId);
    }
    
    @Override
    public Flux<MessageResponse> searchMessages(String keyword, Long userId, Pageable pageable) {
        return messageRepository.searchMessages(keyword, pageable)
                .map(this::convertToMessageResponse);
    }
    
    @Override
    public Mono<MessageResponse> forwardMessage(String originalMessageId, Long fromUserId, Long toUserId, Long groupId) {
        return messageRepository.findByMessageId(originalMessageId)
                .switchIfEmpty(Mono.error(new RuntimeException("原消息不存在")))
                .flatMap(originalMessage -> {
                    // 创建转发消息
                    Message forwardMessage = new Message();
                    forwardMessage.setMessageId(IdGenerator.generateMessageId());
                    forwardMessage.setFromUserId(fromUserId);
                    forwardMessage.setToUserId(toUserId);
                    forwardMessage.setGroupId(groupId);
                    forwardMessage.setMessageType(originalMessage.getMessageType());
                    forwardMessage.setContent(originalMessage.getContent());
                    forwardMessage.setStatus(Message.MessageStatus.NORMAL);
                    forwardMessage.setCreatedAt(LocalDateTime.now());
                    forwardMessage.setUpdatedAt(LocalDateTime.now());
                    forwardMessage.setSequence(sequenceGenerator.incrementAndGet());
                    
                    // 设置转发信息
                    Message.MessageExtra extra = new Message.MessageExtra();
                    Message.ForwardInfo forwardInfo = new Message.ForwardInfo();
                    forwardInfo.setOriginalMessageId(originalMessageId);
                    forwardInfo.setOriginalSenderId(originalMessage.getFromUserId());
                    forwardInfo.setForwardTimestamp(LocalDateTime.now());
                    extra.setForwards(List.of(forwardInfo));
                    forwardMessage.setExtra(extra);
                    
                    // 生成会话ID
                    String conversationId;
                    if (groupId != null) {
                        conversationId = conversationService.generateConversationId(
                            com.reactim.message.entity.Conversation.ConversationType.GROUP, groupId);
                    } else {
                        conversationId = conversationService.generateConversationId(
                            com.reactim.message.entity.Conversation.ConversationType.PRIVATE, fromUserId, toUserId);
                    }
                    forwardMessage.setConversationId(conversationId);
                    
                    return messageRepository.save(forwardMessage);
                })
                .map(this::convertToMessageResponse)
                .doOnSuccess(response -> log.info("消息转发成功: {}", response.getMessageId()));
    }
    
    @Override
    public Mono<List<Long>> getMessageReadUsers(String messageId) {
        return messageReadStatusRepository.findByMessageId(messageId)
                .map(readStatus -> readStatus.getReadStatus().stream()
                        .map(MessageReadStatus.ReadInfo::getUserId)
                        .collect(Collectors.toList()))
                .defaultIfEmpty(List.of());
    }
    
    @Override
    public Flux<MessageResponse> getMessagesAfter(String conversationId, LocalDateTime afterTime) {
        return messageRepository.findByConversationIdAndCreatedAtAfterOrderByCreatedAtAsc(conversationId, afterTime)
                .map(this::convertToMessageResponse);
    }
    
    @Override
    public Flux<MessageResponse> getMessagesAfterSequence(String conversationId, Long sequence) {
        return messageRepository.findByConversationIdAndSequenceGreaterThanOrderBySequenceAsc(conversationId, sequence)
                .map(this::convertToMessageResponse);
    }
    
    /**
     * 创建新的读取状态记录
     */
    private Mono<MessageReadStatus> createNewReadStatus(String messageId) {
        MessageReadStatus readStatus = new MessageReadStatus();
        readStatus.setMessageId(messageId);
        readStatus.setReadStatus(List.of());
        readStatus.setCreatedAt(LocalDateTime.now());
        return messageReadStatusRepository.save(readStatus);
    }
    
    /**
     * 生成消息预览内容
     */
    private String generateMessagePreview(Message message) {
        switch (message.getMessageType()) {
            case TEXT:
                return message.getContent().getText() != null ? 
                       message.getContent().getText().getText() : "[文本消息]";
            case IMAGE:
                return "[图片]";
            case VOICE:
                return "[语音]";
            case VIDEO:
                return "[视频]";
            case FILE:
                return "[文件]";
            case LOCATION:
                return "[位置]";
            case EMOJI:
                return "[表情]";
            case CONTACT:
                return "[联系人]";
            case URL_CARD:
                return "[链接]";
            case SYSTEM:
                return message.getContent().getSystem() != null ?
                       message.getContent().getSystem().getContent() : "[系统消息]";
            default:
                return "[消息]";
        }
    }
    
    /**
     * 推送消息给接收方
     */
    private Mono<Void> pushMessageToReceivers(MessageResponse messageResponse) {
        if (messageResponse.getGroupId() != null) {
            // 群聊消息：推送给群组所有成员（除了发送者）
            return messagePushService.pushToGroup(
                messageResponse.getGroupId(), 
                messageResponse, 
                messageResponse.getFromUserId()
            );
        } else if (messageResponse.getToUserId() != null) {
            // 私聊消息：推送给接收者
            return messagePushService.pushToUser(messageResponse.getToUserId(), messageResponse);
        } else {
            // 系统消息或其他类型，暂不处理
            return Mono.empty();
        }
    }
    
    /**
     * 转换Message实体为MessageResponse DTO
     */
    private MessageResponse convertToMessageResponse(Message message) {
        MessageResponse response = new MessageResponse();
        response.setMessageId(message.getMessageId());
        response.setFromUserId(message.getFromUserId());
        response.setToUserId(message.getToUserId());
        response.setGroupId(message.getGroupId());
        response.setConversationId(message.getConversationId());
        response.setMessageType(message.getMessageType());
        response.setContent(message.getContent());
        response.setExtra(message.getExtra());
        response.setStatus(message.getStatus());
        response.setCreatedAt(message.getCreatedAt());
        response.setUpdatedAt(message.getUpdatedAt());
        response.setSequence(message.getSequence());
        return response;
    }
}